
use std::env::var;
use std::thread::spawn;

use clap::Parser;
use tokio_tungstenite::tungstenite::Message;
use tokio_tungstenite::{accept_async, WebSocketStream};
use wtask_base::encrypt::{decrypt, decrypt_str_to_str, WTASK_KEY};
use wtask_base::error::{WError, WResult};
use wtask_base::export::{KillSender, TcpStream, json, kill_channel, log_debug, log_error, log_info, select, serde_from_str, serde_to_string, yield_now};
use wtask_base::machine::{runtime_init, CMDArgsConfigMachine, WSpawn};
use wtask_base::tcp::{stream_read_raw, stream_write_raw, tcp_listen, WTcpStream};
use wtask_app::{AppRoom, SinkExt, SplitSink, SplitStream, StreamExt, cli_app_stream};
use wtask_file::{IncomingBody, Request, Response, StatusCode, header};

use crate::basic::{parse_body_string, response_body, ResponseBody, WEB_SESSION};


#[derive(Debug, Default)]
pub(crate) struct RoomStorage {
    // websocket killer
    pub(crate) ws_killer: Option<KillSender<bool>>,
}

impl RoomStorage {
    pub(crate) fn drop_data(&mut self) {
        if let Some(killer) = self.ws_killer.take() {
            let _ = killer.send(true);
        }
    }
}


struct WsRoom {
    ws: WebSocketStream<TcpStream>,
}

async fn socketstream_next_inner(mut rx: SplitStream<WebSocketStream<TcpStream>>, stream_write: &TcpStream, key: &str) -> WResult<()> {
    loop {
        let result = rx.next().await;
        if result.is_none() {
            log_debug!("ws get no data");
            break;
        }
        let result = result.unwrap();
        if result.is_err() {
            log_debug!("ws data error: {}", result.err().unwrap());
            break;
        }
        let result = result.unwrap();
        if !result.is_binary() {
            log_debug!("ws data not binary");
            break;
        }
        let result = result.into_data();
        let t = stream_write_raw(stream_write, &result, key).await;
        if t.is_err() {
            log_debug!("Write stream error, stop: {:?}", t.err().unwrap());
            break;
        }
        yield_now().await;
    }
    Ok(())
}

async fn socketstream_read_inner(mut tx: SplitSink<WebSocketStream<TcpStream>, Message>, stream_read: &TcpStream, key: &str) -> WResult<()> {
    loop {
        match stream_read_raw(stream_read, key).await {
            Ok(data) => {
                let data = Message::Binary(data.into());
                let result = tx.send(data).await;
                if let Err(e) = result {
                    log_debug!("Send ws error: {e}");
                    break;
                }
                yield_now().await;
            },
            Err(e) => {
                log_debug!("Read room error, stop: {}", e);
                break;
            }
        }
    }
    Ok(())
}

impl WsRoom {
    async fn ws_room_exchange_data(self, key: &str, stream: WTcpStream) -> WResult<()> {
        // log_debug!("Room start change data ...");
        let (stream_read, stream_write) = stream;
        let (tx, rx) = self.ws.split();
        select! {
            _ = socketstream_next_inner(
                rx,
                stream_write.as_ref(),
                key
            ) => {},
            _ = socketstream_read_inner(
                tx,
                stream_read.as_ref(),
                key
            ) => {},
        }
        Ok(())
    }
}



pub(crate) async fn ws_conn(stream: TcpStream) -> WResult<()> {
    let ws_stream = accept_async(stream).await;
    if ws_stream.is_err() {
        log_error!("websocket con error: {}", ws_stream.err().unwrap());
        return Err(WError::NetError("websocket con error".to_owned()));
    }
    let mut f = WsRoom { ws: ws_stream.unwrap() };
    let data = f.ws.next().await;
    if data.is_none() {
        log_debug!("session get data msg");
        return Ok(());
    }
    let data = data.unwrap();
    if data.is_err() {
        log_debug!("session get data error: {}", data.err().unwrap());
        return Ok(());
    }
    let data = data.unwrap();
    if !data.is_binary() {
        log_debug!("session get data type error");
        return Ok(());
    }
    let data = decrypt(&data.into_data(), WTASK_KEY);
    if data.is_err() {
        log_debug!("session get data decrypt error: {}", data.err().unwrap());
        return Ok(());
    }
    let data = data.unwrap();
    let data = String::from_utf8_lossy(&data).to_string();
    let data = data.split_once("\n\n\n");
    if data.is_none() {
        log_debug!("session get data fmt error");
        return Ok(());
    }
    let (app, config) = data.unwrap();
    match (
        AppRoom::try_parse_from(app.split(" ")),
        serde_from_str::<CMDArgsConfigMachine>(config),
    ) {
        (Ok(app), Ok(config)) => {
            log_debug!("session start people: {app:?}, {config:?}");
            let (stream, key) = cli_app_stream(app, config).await?;
            let v = f.ws_room_exchange_data(&key, stream).await;
            if let Err(e) = v {
                log_debug!("session people err stop: {e:?}");
            }
        },
        (a, b) => {
            log_debug!("session parse data error: {:?}, {:?}, {:?}, {:?}", a.err(), b.err(), app, config);
        }
    }
    Ok(())
}


pub(crate) async fn response_room(req: Request<IncomingBody>, token: &str, port: u32) -> WResult<Response<ResponseBody>> {
    let req_status = req.headers().get("Status").ok_or(WError::Empty)?;
    let machine_id = req.headers().get("ID").ok_or(WError::Empty)?;
    let machine_id = decrypt_str_to_str(machine_id.to_str().unwrap(), token)?;

    let mut machine_config = None;
    let mut machine_keys = Vec::new();
    // check login
    if let Some(machines) = WEB_SESSION.login.read().await.as_ref().get(token) {
        machine_keys.extend(machines.keys().map(| v | v.to_string()));
        if let Some(config_temp) = machines.get(&machine_id) {
            machine_config = Some(serde_to_string(&config_temp.config)?);
        }
    };
    if machine_config.is_none() {
        let json = json!({
            "msg": "登录失败",
            "data": {
                "req": machine_id,
                "all": machine_keys
            },
        }).to_string();
        let res = Response::builder()
            .status(StatusCode::UNAUTHORIZED)
            .header(header::CONTENT_TYPE, "application/json")
            .body(response_body(&json, token)).unwrap();
        return Ok(res);
    }
    

    // status manage
    let machine_config = machine_config.unwrap();
    let req_status = req_status.to_str().unwrap().to_lowercase().eq("true");
    let has_room = WEB_SESSION.room.read().await.ws_killer.is_some();
    let port_gap = var("WTASK_WEB_WS_PORT_GAP").unwrap_or("0".to_owned()).parse::<u32>().unwrap_or(0);
    let res_value = match (req_status, has_room) {
        (false, true) => {
            if let Some(killer) = WEB_SESSION.room.write().await.ws_killer.take() {
                let _ = killer.send(true);
            }
            json!({
                "msg": "对话服务关闭成功",
                "data": "",
            })
        },
        (false, false) => {
            json!({
                "msg": "对话服务已关闭",
                "data": "",
            })
        },
        (true, false) => {
            // app
            let app = {
                let body_de = parse_body_string(req, token).await?;
                AppRoom::try_parse_from(body_de.split(" "))
            };
            // socket server
            let listener = tcp_listen(port, "WebSocket");
            if listener.is_err() || app.is_err() {
                let json = json!({
                    "msg": "对话服务启动失败",
                    "data": format!("{:?}, {:?}", listener.err(), app.err()),
                }).to_string();
                let res = Response::builder()
                    .status(StatusCode::BAD_REQUEST)
                    .header(header::CONTENT_TYPE, "application/json")
                    .body(response_body(&json, token)).unwrap();
                return Ok(res);
            }
            let listener = listener.unwrap();
            let (kill_tx, mut kill_rx_parent) = kill_channel(false);
            let kill_tx_clone = kill_tx.clone();
            spawn(move || {
                let runtime = runtime_init(8)?;
                runtime.block_on(async move {
                    log_info!("Start websocket listener {port} ...");
                    let mut wspawn = WSpawn::default();
                    let (_t, mut kill_rx) = wspawn.catch_pause_signal(async move {});
                    loop {
                        select! {
                            Ok((stream, _)) = listener.accept() => {
                                log_debug!("websocket start a connect ...");
                                wspawn.spawn(async move {
                                    let _ = ws_conn(stream).await;
                                });
                            }
                            _ = kill_rx.changed() => {
                                log_info!("Shutting websocket {port} ...");
                                break;
                            }
                            _ = kill_rx_parent.changed() => {
                                log_info!("Shutting websocket {port}, kill by parent ...");
                                let _ = kill_tx_clone.send(true);
                                break;
                            }
                        }
                    }
                });
                runtime.shutdown_background();
                Ok::<_, WError>(())
            });
            WEB_SESSION.room.write().await.ws_killer.replace(kill_tx);
            json!({
                "msg": "对话服务启动成功",
                "data": {
                    "port": port + port_gap,
                    "config": machine_config
                },
            })
        },
        (true, true) => {
            json!({
                "msg": "对话服务已启动",
                "data": {
                    "port": port + port_gap,
                    "config": machine_config
                },
            })
        }
    };

    let json = json!(res_value).to_string();
    let res = Response::builder()
        .status(StatusCode::OK)
        .header(header::CONTENT_TYPE, "application/json")
        .body(response_body(&json, token)).unwrap();
    Ok(res)
}

